package com.lb.bs.client.zookeeper;

import com.alibaba.fastjson.JSON;
import com.lb.bs.client.beans.LBGrayIPBean;
import com.lb.bs.client.beans.LBGrayItemBean;
import com.lb.bs.client.beans.LBItemBean;
import com.lb.bs.client.repository.StoreCenter;
import com.lb.bs.client.factory.SingletonFactory;
import com.lb.bs.client.processor.beans.ProcessorBean;
import com.lb.bs.client.processor.beans.ProcessorDynamicBean;
import com.lb.bs.client.processor.enums.ProcessorType;
import com.lb.bs.client.processor.mgr.ProcessorMgr;
import com.lb.bs.client.serialize.SerializationHolder;
import com.lb.bs.client.serialize.SerializationInterface;
import com.lb.bs.client.util.StringUtils;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.ChildData;
import org.apache.curator.framework.recipes.cache.TreeCache;
import org.apache.curator.framework.recipes.cache.TreeCacheEvent;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * program: LB-BS-Client
 * author: bsworld.xie
 * create: 2019-11-27 17:46
 * description:
 */
public class WatchMgr {
    private static final Logger logger = LoggerFactory.getLogger(WatchMgr.class);

    ExecutorService executorService = Executors.newSingleThreadExecutor();
    private int flag = 0;

    public void watch(final String key) {
        logger.debug("watch(),init, key:{}", key);
        try {
            CuratorFramework client = ZKMgr.getInstance().getClient();
            ProcessorMgr processorMgr = SingletonFactory.getInstance(ProcessorMgr.class);
            TreeCache treeCache = new TreeCache(client, key);
            treeCache.getListenable().addListener((client1, event) -> {
                ProcessorDynamicBean dynamicBean = null;
                try {
                    boolean check = check(event);
                    if (!check) {
                        return;
                    }
                    dynamicBean = wrapperChangeData(event);
                    ProcessorBean.Builder builder = ProcessorBean.builder().data(dynamicBean).async(true);
                    switch (event.getType()) {
                        case NODE_ADDED:
                            processorMgr.route(builder.processorType(ProcessorType.CREATE).build());
                            break;
                        case NODE_UPDATED:
                            processorMgr.route(builder.processorType(ProcessorType.UPDATE).build());
                            break;
                        case NODE_REMOVED:
                            processorMgr.route(builder.processorType(ProcessorType.REMOVE).build());
                            break;
//                    //通过对session expired时间测试，发生该事件后,通过测试发现,在重新连接后会把
//                    // 过期时间内发生的数据重新回调,不用对session expired事件做特殊处理
//                    case CONNECTION_LOST:
//                    case CONNECTION_RECONNECTED:
//                    case CONNECTION_SUSPENDED:
//                        logger.info("watch(), special event:{}", event);
//
//                        //对于connection lost可以不用关注
//                        // 创建客户端时的链接就已经加上了重试策略，该状态的产生只是代表重试
//                        //参考链接:http://curator.apache.org/errors.html
//                        break;
                    }
                } catch (Throwable t) {
                    logger.error("watch callback exception, event:{}, dynamicBean:{},error:", JSON.toJSONString(event), JSON.toJSONString(dynamicBean),t);
                }


            }, executorService);
            treeCache.start();
        } catch (Exception e) {
            logger.info("watchOneItem(),key:{}, error:", key, e);
        }
    }


    private boolean check(TreeCacheEvent event) {
        boolean checkEvent = checkEvent(event);
        if (!checkEvent) {
            return false;
        }
        String zkKey = event.getData().getPath();
        boolean inWatchKey = checkInWatchKey(zkKey);
        logger.info("WatchMgr.class, watch(), zkKey:{}, event:{}", zkKey, event);
        if (!inWatchKey) {
            logger.warn("WatchMgr.class, watch(), zkKey:{}, not in watch", zkKey);
            return false;
        }
        return true;
    }

    private boolean checkEvent(TreeCacheEvent event) {
        ChildData data = event.getData();
        if (data == null) {
            return false;
        }
        String path = data.getPath();
        if (StringUtils.isEmpty(path)) {
            return false;
        }
        TreeCacheEvent.Type type = event.getType();
        if (type == null) {
            return false;
        }
        return true;
    }

    private boolean checkInWatchKey(String zkKey) {
        StoreCenter storeCenter = StoreCenter.getInstance();
        Map<String, LBItemBean> itemKeyMap = storeCenter.getItemKeyMap();
        Map<String, LBGrayIPBean> grayIPKeyMap = storeCenter.getGrayIPKeyMap();
        Map<String, LBGrayItemBean> grayItemKeyMap = storeCenter.getGrayItemKeyMap();

        if (itemKeyMap.containsKey(zkKey)) {
            return true;
        }

        if (grayIPKeyMap.containsKey(zkKey)) {
            return true;
        }

        if (grayItemKeyMap.containsKey(zkKey)) {
            return true;
        }

        return false;
    }


    public ProcessorDynamicBean wrapperChangeData(TreeCacheEvent event) throws Exception {
        ChildData data = event.getData();
        byte[] zkData = data.getData();
        String watchItemKey = data.getPath();
        String zkDataStr = null;
        if (zkData != null) {
            SerializationInterface holder = SerializationHolder.getHolder();
            zkDataStr = holder.deserialize(zkData);
        }
        TreeCacheEvent.Type type = event.getType();
        ProcessorDynamicBean.Builder builder = ProcessorDynamicBean.builder().key(watchItemKey).data(zkDataStr);
        if (TreeCacheEvent.Type.NODE_UPDATED.equals(type)) {
            builder.enableUpdateCallBack(true);
        }
        ProcessorDynamicBean dynamicBean = builder.build();
        return dynamicBean;

    }


    public static void main(String[] args) throws Exception {
//        Socket socket = new Socket("106.13.46.179", 2181);
//        while (true) {
//            System.out.println(socket);
//        }


//        String connectString = "127.0.0.1:2181";
//
//        String watchPath = "/lb";
//
//        ZooKeeper zooKeeper = new ZooKeeper(connectString, 1000000,null);
//        TimeUnit.SECONDS.sleep(5);
////        String s = zooKeeper.create("/hello", null, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
////        System.out.println(s);
//        zooKeeper.exists("/hello", true);
//        System.in.read();


    }

}
